// client_writer.go
//
// 功能：演示 SafeWriter 让多个生产者 goroutine 安全地向同一 socket 写入。
// 步骤：
//   1. 建立 TCP 连接并封装 SafeWriter。
//   2. 同时启动多个 goroutine，通过 SafeWriter 的 SubmitWithAck 顺序写入消息。
//   3. 主协程持续读取服务器响应，观察写入顺序保持一致。
//   4. 所有写入完成后，优雅关闭 SafeWriter 和连接。

package main

import (
	"bufio"
	"errors"
	"fmt"
	"log"
	"net"
	"strings"
	"sync"
	"time"
)

const writerServerAddr = "127.0.0.1:8100"

// WriteRequest 表示一次写入请求，可携带可选的结果回传通道。
type WriteRequest struct {
	Data   []byte
	Result chan error // 若不需要回传，可为 nil
}

// SafeWriter 将多个 goroutine 的写入集中到单一协程串行写 socket。
type SafeWriter struct {
	conn      net.Conn
	queue     chan WriteRequest
	closeOnce sync.Once
	done      chan struct{}
}

// NewSafeWriter 创建 SafeWriter，并启动内部写协程。
func NewSafeWriter(conn net.Conn, queueSize int) *SafeWriter {
	sw := &SafeWriter{
		conn:  conn,
		queue: make(chan WriteRequest, queueSize),
		done:  make(chan struct{}),
	}
	go sw.loop()
	return sw
}

// loop 是真正写 socket 的协程，顺序消费队列中的请求。
func (sw *SafeWriter) loop() {
	for req := range sw.queue {
		_, err := sw.conn.Write(req.Data)
		if req.Result != nil {
			req.Result <- err
		}
	}
	close(sw.done)
}

// Submit 提交写入（无需关心结果）。
func (sw *SafeWriter) Submit(data []byte) error {
	return sw.SubmitWithAck(data, nil)
}

// SubmitWithAck 提交写入并在 Result 通道返回写入是否成功。
func (sw *SafeWriter) SubmitWithAck(data []byte, result chan error) error {
	if result == nil {
		result = make(chan error, 1) // 保证 loop 写入时不会阻塞
	}
	select {
	case sw.queue <- WriteRequest{Data: data, Result: result}:
	case <-sw.done:
		return fmt.Errorf("SafeWriter 已关闭")
	}
	return <-result
}

// Close 关闭 SafeWriter，等待写协程退出后再关闭底层连接。
func (sw *SafeWriter) Close() error {
	var err error
	sw.closeOnce.Do(func() {
		close(sw.queue)
		<-sw.done
		err = sw.conn.Close()
	})
	return err
}

func main() {
	conn, err := net.Dial("tcp", writerServerAddr)
	if err != nil {
		log.Fatalf("客户端拨号失败: %v", err)
	}
	// 注意：取消之前的 defer conn.Close()，因为我们要手动控制关闭顺序。

	sw := NewSafeWriter(conn, 32)

	var writerWG sync.WaitGroup
	writerCount := 5
	messagePerWriter := 3
	for i := 0; i < writerCount; i++ {
		writerID := i
		writerWG.Add(1)
		go func() {
			defer writerWG.Done()
			for j := 1; j <= messagePerWriter; j++ {
				payload := fmt.Sprintf("写协程-%d 的消息-%d\n", writerID, j)
				if err := sw.SubmitWithAck([]byte(payload), nil); err != nil {
					log.Printf("[writer-client] SafeWriter 写入失败: %v", err)
					return
				}
				time.Sleep(100 * time.Millisecond)
			}
		}()
	}

	var readerWG sync.WaitGroup
	readerWG.Add(1)
	go func() {
		defer readerWG.Done()
		reader := bufio.NewScanner(conn)
		for reader.Scan() {
			log.Printf("[writer-client] 收到回显: %s", reader.Text())
		}
		if err := reader.Err(); err != nil {
			if errors.Is(err, net.ErrClosed) || isUseOfClosedErr(err) {
				log.Println("[writer-client] 读取协程退出：连接已关闭（预期）")
			} else {
				log.Printf("[writer-client] 读取服务器出错: %v", err)
			}
		} else {
			log.Println("[writer-client] 服务器正常关闭连接")
		}
	}()

	writerWG.Wait() // 等待所有写协程结束
	log.Println("[writer-client] 写入全部完成，准备关闭 SafeWriter")

	sw.Close()      // 停止写入并关闭连接
	readerWG.Wait() // 等待读取协程处理完关闭事件
	log.Println("[writer-client] 客户端退出")
}

// isUseOfClosedErr 针对 “use of closed network connection” 的兼容判断。
func isUseOfClosedErr(err error) bool {
	// Go 1.20+ 会把它包装在 *net.OpError 中，这里做字符串兼容。
	return strings.Contains(err.Error(), "use of closed network connection")
}
